package org.jfsm.concurrent;

import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReferenceArray;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

import org.jfsm.log.LogSource;

public final class ArrayBlockingQueue<T> {

    LogSource logger = new LogSource();

    private final AtomicInteger readPos = new AtomicInteger();
    private final AtomicInteger writePos = new AtomicInteger();
    private final int posMask;

    private final AtomicReferenceArray<Object> queue;

    private static final Object waitNode = new Object();
    private static final Object notifyNode = new Object();

    private final ReentrantLock lock = new ReentrantLock();
    private final Condition notEmpty = lock.newCondition();
    
    /**
     * Constructor
     * 
     * @param minCapacity
     *            Choose 2^n
     */
    public ArrayBlockingQueue(int minCapacity) {
        int qBits, qSize;
        for (qBits = 2, qSize = 4; qSize < minCapacity; qBits++) {
            qSize <<= 1;
        }

        posMask = qSize - 1;
        queue = new AtomicReferenceArray<Object>(qSize);
        queue.set(0, waitNode);
    }

    public final boolean offer(T item) {
        //int wid = worker.get();
        for (;;) {
            int pos = writePos.get();
            int nextPos = (pos+1) & posMask;
            //logger.log(wid+" offer pos:", pos);
            Object nextNode;
            if (queue.compareAndSet(nextPos, null, waitNode)) {
                nextNode = waitNode;
            }
            else {
                nextNode = queue.get(nextPos);
            }
            
            for(;;) {
                Object node = queue.get(pos);
                if (node == waitNode || node == notifyNode) {
                    if( queue.compareAndSet(pos, node, item)) {
                        writePos.set(nextPos);
                        if (nextNode == waitNode) {
                            if (node == waitNode) {
                                return true;
                            }
                            else { //if(node == notifyNode)
                                lock.lock();
                                //logger.log(wid+" offer notify pos:", pos);
                                notEmpty.signalAll();
                                lock.unlock();
                                return true;
                            }
                        }
                        else {
                            //logger.log(wid+" queue overflow pos:", pos);
                            return false;
                        }
                    }
                    else {
                        //logger.log(wid+" offer race pos:", pos);                        
                    }
                }
                else {
                    //logger.log(wid+" offer skip pos:", pos);
                    break;
                }
            }
        }
    }

    /* For Debug
    private static final ThreadLocal<Integer> worker = new ThreadLocal<Integer>() {
        AtomicInteger id = new AtomicInteger();
        protected Integer initialValue() {return id.getAndIncrement();};
    };
    */
    
    @SuppressWarnings("unchecked")
    public final T take() throws ConcurrentException {
        //int wid = worker.get();
        for (int pos = readPos.get(), nextPos = (pos+1) & posMask; ; ) {
            //logger.log(wid+" take pos:", pos);
            
            Object item = queue.get(pos);
            if (item != null) {
                if (item != waitNode && item != notifyNode) {
                    if (queue.compareAndSet(pos, item, null)) {
                        readPos.set(nextPos);
                        return (T) item;
                    } else {
                        //logger.log(wid+" take race pos:", pos);
                    }
                }
                else {
                    //logger.log(wid+" take yield pos:", pos);
                    Thread.yield();
                    pos = readPos.get();
                    nextPos = (pos+1) & posMask;
                    item = queue.get(pos);
                    
                    if (item != null) {
                        if (item != notifyNode) {
                            if (item != waitNode) {
                                if (queue.compareAndSet(pos, item, null)) {
                                    readPos.set(nextPos);
                                    return (T) item;
                                } else {
                                    //logger.log(wid+" take race pos:", pos);
                                    pos = nextPos;                    
                                    nextPos = (pos+1) & posMask;
                                    continue;
                                }
                            }
                            else if (!queue.compareAndSet(pos, item, notifyNode)) {
                                continue;
                            }
                        }
                        
                        lock.lock();
                        //logger.log(wid+" take wait pos:", pos);
                        if( queue.get(pos) == notifyNode ) {
                            notEmpty.awaitUninterruptibly();
                        }
                        lock.unlock();
                        pos = readPos.get();
                        nextPos = (pos+1) & posMask;
                        continue;
                    }
                }
            }
            else {
                //logger.log(wid+" take skip pos:", pos);
            }
            pos = nextPos;                    
            nextPos = (pos+1) & posMask;
        }
    }
}
